package com.atguigu.gmall.realtime.app.dwd;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.util.DateFormatUtil;
import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import com.atguigu.gmall.realtime.util.WasteUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class DwdTrafficSplitApp {


    //目前用户行为数据都集中在topic_log
    // 1  附加任务：  校正埋点标记的is_new      算子状态？ 要用键控状态 （数据一致性，数据量大)
    //
    //2 通过一个程序把topic_log 的数据拆分成多条流 ，以便下游统计分析使用。sideoutput
    //
    //3 分流后的数据写入到kafka中
    public static void main(String[] args) throws Exception {

        //  --waste true
        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        String waste = parameterTool.get("waste");


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
      //  env.setParallelism(4);
        //1  抓取数据
        String topic="topic_log";
        String groupId="dwd_traffic_split_app";
        DataStreamSource<String> kafkaDstream = env.addSource(MyKafkaUtil.getKafkaConsumer(topic, groupId)).setParallelism(1);

    //    env.setStateBackend(new HashMapStateBackend());
    //    env.setStateBackend(new EmbeddedRocksDBStateBackend(true));


        //2 转换结构，转换不成的脏数据做侧输出
        OutputTag<String> dirtyDataOutputTag=new OutputTag<String>("dirty_data"){};
        SingleOutputStreamOperator<JSONObject> jsonObjDstream = kafkaDstream.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String jsonString, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(jsonString);
                    out.collect(jsonObject);   // 如果json转换不成功，脏数据，
                } catch (Exception e) {
                    e.printStackTrace();
                    ctx.output(dirtyDataOutputTag, jsonString);
                }
            }
        }).uid("convert_json").name("convert_json");

        //把脏数据分流出来
        DataStream<String> dirtyDataStream = jsonObjDstream.getSideOutput(dirtyDataOutputTag);
        dirtyDataStream.print("dirty_data:");


       //3  附加任务：  校正埋点标记的is_new      算子状态？ 要用键控状态 （数据一致性，数据量大)
        // is_new 可能1 可能0   如果是0 标识埋点认为老用户不用校正   如果是1  有可能是用户清楚了缓存 需要校正
        //  3.1 如果是1:        保存 用户首次访问日期到 状态中
        //             把用户访问的记录的时间戳和  【用户首次访问日期】 进行比较
        //            情况1  ： 【用户首次访问日期】 不存在    新用户   把当前访问的日期作为【用户首次访问日期】 保存到状态中
                     //情况2 ：用户访问的记录的时间戳 和  【用户首次访问日期】 相同 新用户
                   //  情况3  用户访问的记录的时间戳 大于  【用户首次访问日期】
        //  3.2  如果是0 ：
        //       埋点标记为老用户 ，但是状态中不存在
        //       以埋点为准 老用户  ，补充状态【用户首次访问日期】 ，放前一天的日期  更新状态


        KeyedStream<JSONObject, String> keyedMidJsonObjDstream = jsonObjDstream.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));


        SingleOutputStreamOperator<JSONObject> checkedIsNewJsonObjDstream = keyedMidJsonObjDstream.map(new RichMapFunction<JSONObject, JSONObject>() {
            //状态  键控状态
            ValueStateDescriptor<String> firstVisitDateStateDesc = new ValueStateDescriptor<String>("first_visit_date_state", String.class);

            ValueState<String> firstVisitDateState = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                firstVisitDateState = getRuntimeContext().getState(firstVisitDateStateDesc);
            }

            @Override
            public JSONObject map(JSONObject jsonObject) throws Exception {
                String isNew = jsonObject.getJSONObject("common").getString("is_new");
                Long ts = jsonObject.getLong("ts");
                String curDate = DateFormatUtil.toDate(ts);


                //
                if("true".equals(waste)){
                    WasteUtil.waste(6);
                }


                if("true".equals(waste)){
                    WasteUtil.waste(7);
                }


                if (isNew.equals("1")) {
                    //  3.1 如果是1:   保存 用户首次访问日期到 状态中
                    //    把用户访问的记录的时间戳和  【用户首次访问日期】 进行比较
                    if (firstVisitDateState.value() == null || firstVisitDateState.value().length() == 0) {
                        // 情况1  ： 【用户首次访问日期】 不存在    新用户   把当前访问的日期作为【用户首次访问日期】 保存到状态中
                        //把ts 转成 date_str
                        firstVisitDateState.update(curDate);
                    } else {
                        String firstVisitDate = firstVisitDateState.value();
                        if (!curDate.equals(firstVisitDate)) {
                            //  情况3  用户访问的记录的时间戳 大于(不相同)  【用户首次访问日期】  校正为老用户
                            jsonObject.getJSONObject("common").put("is_new", "0");
                        }

                    }
                    //情况2 ：用户访问的记录的时间戳 和  【用户首次访问日期】 相同 新用户    什么都不用做
                } else {
                    //  3.2  如果是0 ：
                    //       埋点标记为老用户 ，但是状态中不存在
                    //       以埋点为准 老用户  ，补充状态【用户首次访问日期】 ，放前一天的日期  更新状态
                    if (firstVisitDateState.value() == null || firstVisitDateState.value().length() == 0) {
                        String yesterdayDate = DateFormatUtil.toDate(ts - 1000 * 60 * 60 * 24);
                        firstVisitDateState.update(yesterdayDate);
                    }
                }
                return jsonObject;
            }
        }).uid("is_check").name("is_check");


        checkedIsNewJsonObjDstream.print(">>>");


        //2 通过一个程序把topic_log 的数据拆分成多条流 ，以便下游统计分析使用。sideoutput
        //   start   ->   有start 元素   common  ts  start    侧输出
        //   page    ->    有page 元素   common ts  page      主流
        //   action  -> 有page 元素  , actions 数组  拆分 1对多     common  page   +一个action    侧输出
        //   display -> 有page 元素  , displays 数组  拆分 1对多    common  page  +一个diplay    侧输出
        //   err  ->    有err 元素  维持 jsonobj 侧输出

        //定义侧输出的tag  4个
        OutputTag<String> startOutputTag=new OutputTag<String>("start"){};
        OutputTag<String> actionOutputTag=new OutputTag<String>("action"){};
        OutputTag<String> displayOutputTag=new OutputTag<String>("display"){};
        OutputTag<String> errOutputTag=new OutputTag<String>("err"){};


        SingleOutputStreamOperator<String> splitedDstream = checkedIsNewJsonObjDstream.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, String>.Context ctx, Collector<String> out) throws Exception {

                JSONObject errJsonObj = jsonObject.getJSONObject("err");
                if (errJsonObj != null && errJsonObj.size() > 0) {             //   err  ->    有err 元素  维持 jsonobj 侧输出
                    ctx.output(errOutputTag, jsonObject.toJSONString());
                } else {
                    //   start   ->   有start 元素   common  ts  start    侧输出

                    JSONObject startJsonObj = jsonObject.getJSONObject("start");
                    if (startJsonObj != null && startJsonObj.size() > 0) {
                        ctx.output(startOutputTag, jsonObject.toJSONString());
                    } else {
                        JSONObject pageJsonObj = jsonObject.getJSONObject("page");
                        JSONObject commonJsonObj = jsonObject.getJSONObject("common");

                        //   action  -> 有page 元素  , actions 数组  拆分 1对多     common  page   +一个action    侧输出
                        JSONArray actionJsonArr = jsonObject.getJSONArray("actions");
                        if (actionJsonArr != null && actionJsonArr.size() > 0) {
                            for (int i = 0; i < actionJsonArr.size(); i++) {
                                JSONObject actionJsonObj = actionJsonArr.getJSONObject(i);
                                JSONObject newActionJSONobj = new JSONObject();  // common  page   +一个action
                                newActionJSONobj.put("page", pageJsonObj);
                                newActionJSONobj.put("common", commonJsonObj);
                                newActionJSONobj.put("action", actionJsonObj);
                                ctx.output(actionOutputTag, newActionJSONobj.toJSONString());
                            }
                        }

                        //   display -> 有page 元素  , displays 数组  拆分 1对多    common  page  +一个diplay    侧输出
                        JSONArray displayJsonArr = jsonObject.getJSONArray("displays");
                        if (displayJsonArr != null && displayJsonArr.size() > 0) {
                            for (int i = 0; i < displayJsonArr.size(); i++) {
                                JSONObject displayJsonObj = displayJsonArr.getJSONObject(i);
                                JSONObject newDisplayJSONobj = new JSONObject();  // common  page   +一个action
                                newDisplayJSONobj.put("page", pageJsonObj);
                                newDisplayJSONobj.put("common", commonJsonObj);
                                newDisplayJSONobj.put("display", displayJsonObj);
                                newDisplayJSONobj.put("ts", jsonObject.getLong("ts"));
                                ctx.output(displayOutputTag, newDisplayJSONobj.toJSONString());
                            }
                        }

                        //   page    ->    有page 元素   common ts  page      主流
                        jsonObject.remove("displays");
                        jsonObject.remove("actions");
                        out.collect(jsonObject.toJSONString());
                    }
                }
            }
        });

        //根据tag把4个分支侧输出出来
        DataStream<String> errDstream = splitedDstream.getSideOutput(errOutputTag);
        DataStream<String> displayDstream = splitedDstream.getSideOutput(displayOutputTag);
        DataStream<String> actionDstream = splitedDstream.getSideOutput(actionOutputTag);
        DataStream<String> startDstream = splitedDstream.getSideOutput(startOutputTag);


        //打印
        splitedDstream.print("page>>>>>>>");
        errDstream.print("errrrrrrrr>>>>>>>");
        displayDstream.print("dddddddddd>>");
        actionDstream.print("aaaaaaaaa>>>>>>>");
        startDstream.print("sssssssss>>>>>>>");



        splitedDstream.addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_page_log")).uid("sink_page").name("sink_page");
        errDstream.addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_error_log"));
        displayDstream.addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_display_log"));
        actionDstream.addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_action_log"));
        startDstream.addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_start_log"));


        env.execute();


    }
}
